package com.chenjl.api;

import java.util.ArrayList;
import java.util.Collection;
import java.util.HashSet;
import java.util.Set;

import org.apache.kafka.clients.producer.Partitioner;
import org.apache.kafka.clients.producer.internals.DefaultPartitioner;
import org.apache.kafka.common.Cluster;
import org.apache.kafka.common.Node;
import org.apache.kafka.common.PartitionInfo;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
/**
 * 测试Partitioner
 * 2016-9-19 17:41:34
 * @author chenjinlong
 */
public class TestOnPartitioner {
	static Logger logger = LoggerFactory.getLogger(TestOnPartitioner.class);
	
	public static void main(String[] args) {
		Collection<Node> nodes = new ArrayList<Node>();
		Node node01 = new Node(101,"192.168.245.127",9092);
		Node node02 = new Node(102,"192.168.245.128",9092);
		Node node03 = new Node(103,"192.168.245.129",9092);
		nodes.add(node01);
		nodes.add(node02);
		nodes.add(node03);
		Node[] nodeArray = {node01,node02,node03};
		
		Collection<PartitionInfo> partitionInfos = new ArrayList<PartitionInfo>();
		partitionInfos.add(new PartitionInfo("topic01",1,node01,nodeArray,nodeArray));
		partitionInfos.add(new PartitionInfo("topic01",2,node01,nodeArray,nodeArray));
		partitionInfos.add(new PartitionInfo("topic01",0,node01,nodeArray,nodeArray));
		
		Set<String> unauthorizedTopics = new HashSet<String>();
		unauthorizedTopics.add("topic01");
		unauthorizedTopics.add("topic02");
		
		Cluster cluster = new Cluster(nodes,partitionInfos,unauthorizedTopics);
		
		Partitioner partitioner = new DefaultPartitioner();
		for(int i=1;i<=10;i++) {
			//DefaultPartitioner 无keyBytes时，根据PartitionInfo做轮询
			int partition = partitioner.partition("topic01",null,null,null,null,cluster);
			logger.info("round : {} , partition : {}",i,partition);
		}
		
		
		System.out.println("\n");
		for(int i=1;i<=10;i++) {
			//DefaultPartitioner 根据 keyBytes做Hash
			byte[] keyBytes = ("hh"+i).getBytes();
			int partition = partitioner.partition("topic01",null,keyBytes,null,null,cluster);
			logger.info("round : {} , partition : {}",i,partition);
		}
	}
}